package com.atguigu.flink.chapter08;




import com.atguigu.flink.bean.UserBehavior;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Flink02_Project_Product_PV_1 {

       public static void main(String[] args) {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
               env.setParallelism(1);

               env.readTextFile("input/UserBehavior.csv")
                       .map(line->{
                           String[] datas = line.split(",");
                           return new UserBehavior(
                                   Long.valueOf(datas[0]),
                                   Long.valueOf(datas[1]),
                                   Integer.valueOf(datas[2]),
                                   datas[3],
                                   Long.parseLong(datas[4])*1000
                           );
                       })
                       // 添加水印
                       .assignTimestampsAndWatermarks(
                               WatermarkStrategy
                                       .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                       //添加时间戳
                               .withTimestampAssigner((ub,ts)->ub.getTimestamp())
                       )
                       // 过滤行为，只要 pv 行为
                       .filter(ub ->"pv".equals(ub.getBehavior()))
                       .keyBy(UserBehavior::getBehavior)
                       .window(TumblingEventTimeWindows.of(Time.minutes(30)))

                      .process(new ProcessWindowFunction<UserBehavior, String, String, TimeWindow>() {

                          private ReducingState<Long> pvState;

                          @Override
                          public void open(Configuration parameters) throws Exception {
                              pvState = getRuntimeContext()
                                      .getReducingState(
                                      new ReducingStateDescriptor<Long>(
                                      "pvState", Long::sum, Long.class));
                          }

                          @Override
                          public void process(String key,
                                              Context ctx,
                                              Iterable<UserBehavior> elements,
                                              Collector<String> out) throws Exception {

                            //键控状态只与key相关，所以要clear
                              pvState.clear();

                              for (UserBehavior element : elements) {
                                  pvState.add(1L);
                              }
                              out.collect(ctx.window()+ " " +pvState.get());
                          }
                      })
                       .print();






               try {
                   env.execute();
               } catch (Exception e) {
                   e.printStackTrace();
               }


           }


}
